package dict

import java.util.Properties

import ch.hsr.geohash.GeoHash
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

object GeoHashDict {
  def main(args: Array[String]): Unit = {
    // 构造spark
    val session: SparkSession = SparkSession.builder().appName("GeoHashDict").master("local[4]").getOrCreate()
    import session.implicits._
    println(this.getClass.getName)
    // 读取MySQL中的gps坐标地理位置表
    val props: Properties = new Properties()
    props.setProperty("user", "root")
    props.setProperty("password", "123456")
    val frame: DataFrame = session.read.jdbc("jdbc:mysql://localhost:3306/doit_yiee", "tmp", props)
    val res: DataFrame = frame.map(row => {
      val lng: Double = row.getAs[Double]("BD09_LNG")
      val lat: Double = row.getAs[Double]("BD09_LAT")
      val district: String = row.getAs[String]("district")
      val city: String = row.getAs[String]("city")
      val province: String = row.getAs[String]("province")
      // 将GPS坐标信息转为geoHash编码
      val geoCode: String = GeoHash.geoHashStringWithCharacterPrecision(lat, lng, 5)
      // 组装结果
      (geoCode, province, city, district)
    }).toDF("geo", "province", "city", "district")
    //    res.show(10, truncate = false)
    // 保存结果
    res.write.parquet("data/dict/geo_dict/output")
    session.close()
  }
}
